package mqConsumer

import (
	"context"
	"encoding/json"
	"github.com/Sir-yuChen/funAnalyse/global"
	"github.com/Sir-yuChen/funAnalyse/global/constant"
	"github.com/Sir-yuChen/funAnalyse/model"
	"github.com/Sir-yuChen/funAnalyse/utils/kafka"
	"github.com/Sir-yuChen/funAnalyse/utils/kafka/spec"
	"github.com/Sir-yuChen/funAnalyse/utils/mongodb"
	"github.com/Sir-yuChen/funAnalyse/utils/socket"
	"go.mongodb.org/mongo-driver/bson"
	"go.uber.org/zap"
	"time"
)

type CommunicationMsgConsumer struct{}

func (c CommunicationMsgConsumer) Consumer(kmsg spec.Msg) {
	ctx := context.TODO()
	var kaJson kafka.MsgDataRequest
	var completeInformation model.CompleteInformation
	if string(kmsg.Data()) == "" {
		global.GVA_LOG.Error("kafka consumer nil", zap.Any("msgKey", kmsg.GetKey()))
		return
	}
	//1.json转map
	err := json.Unmarshal(kmsg.Data(), &kaJson)
	if err == nil {
		err = json.Unmarshal([]byte(kaJson.Content), &completeInformation)
	}
	if err != nil {
		global.GVA_LOG.Error(" 数据转换异常", zap.Any("kmsg", kmsg.Data()), zap.Error(err))
		return
	}
	filter := bson.M{
		"informationid": completeInformation.BaseInfo.InformationId,
	}
	count, err := mongodb.CommunicationBaseInformationCollection.SelectCount(ctx, filter)
	if err != nil {
		global.GVA_LOG.Error("查消息基础信息异常", zap.Any("completeInformation", completeInformation), zap.Error(err))
		return
	}
	if count == 0 {
		//基础消息落库
		if _, err := mongodb.CommunicationBaseInformationCollection.InsetOne(ctx, completeInformation.BaseInfo); err != nil {
			global.GVA_LOG.Error("新增基础消息记录异常", zap.Any("completeInformation", completeInformation), zap.Error(err))
			return
		}
	}
	filterRomm := bson.M{
		"roomid": completeInformation.BaseInfo.RoomId,
	}
	//查房间中消息接收者
	baseRoomInfo, err := mongodb.CommunicationRoomCollection.SelectOne(ctx, filterRomm)
	if err != nil {
		global.GVA_LOG.Error("查聊天室基础信息异常", zap.Any("RoomId", completeInformation.BaseInfo.RoomId), zap.Error(err))
		return
	}
	var roomInfo model.BaseRoom
	bytes, err := bson.Marshal(baseRoomInfo)
	err = bson.Unmarshal(bytes, &roomInfo)
	if err != nil {
		global.GVA_LOG.Error("数据转换异常", zap.Any("completeInformation", completeInformation), zap.Error(err))
		return
	}
	roomMenmberFilter := bson.M{
		"roomid": roomInfo.RoomId,
		"status": constant.STATUS_Y,
	}
	memberInfos, err := mongodb.CommunicationRoomMemberCollection.SelectList(ctx, roomMenmberFilter, nil)
	if err != nil {
		global.GVA_LOG.Error("查聊天室成员异常", zap.Any("completeInformation", completeInformation), zap.Error(err))
		return
	}
	for _, item := range memberInfos {
		member := new(model.RoomMenber)
		bytes, err := bson.Marshal(item)
		if err != nil {
			global.GVA_LOG.Error("Mongdb 查询集合数据转换异常", zap.Any("jsonStr", item), zap.Error(err))
		}
		bson.Unmarshal(bytes, member)
		information := model.InformationOnUser{
			CustomerId:        member.CustomerId,
			InformationId:     completeInformation.BaseInfo.InformationId,
			UserIdentity:      constant.MSG_User_Identity_recipient,
			InformationStatus: constant.MSG_STATUS_WILL_processed,
			CreateTime:        time.Now().Format("2006-01-02 15:04:05"),
		}
		if completeInformation.InformationSender.CustomerId == member.CustomerId {
			information = completeInformation.InformationSender
		}
		//判断接收者是否在线
		client := socket.GetClient(information.CustomerId)
		for _, cl := range client {
			if cl.UseMark == global.ChatStock {
				var message socket.ServiceMessage
				m := map[string]interface{}{}
				m["msg"] = completeInformation.BaseInfo.InformationContent
				m["userIdentity"] = information.UserIdentity
				marshal, err := json.Marshal(m)
				if err != nil {
					global.GVA_LOG.Error("数据转换异常", zap.Any("completeInformation", completeInformation), zap.Error(err))
					continue
				}
				content := socket.ServiceMessageContent{
					Body: string(marshal),
				}
				message.Type = completeInformation.BaseInfo.InformationType
				message.Content = content
				if err := cl.Send(message); err != nil {
					global.GVA_LOG.Error("消息发送失败", zap.Any("CustomerId：", information.CustomerId), zap.Error(err))
					information.InformationStatus = constant.MSG_STATUS_WILL_Processe_failed
					//todo 优化为重试机制
				} else {
					information.InformationStatus = constant.MSG_STATUS_WILL_Processe_success
				}
			}
		}
		//将发送者接收者相对应消息落库
		if _, err := mongodb.CommInformationOnUserCollection.InsetOne(ctx, information); err != nil {
			global.GVA_LOG.Error("新增用户关联消息记录异常", zap.Any("InformationOnUser", information), zap.Error(err))
			continue
		}
	}
}

func (c CommunicationMsgConsumer) FailAction(err *kafka.ConsumeError) {
	global.GVA_LOG.Error("Kafka 消费消息异常 执行 FailAction", zap.Any("err", err.Err))
}
